package com.broada.apm.indicator.api.util;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.lucene.search.TopDocs;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
import org.elasticsearch.search.aggregations.bucket.filters.FiltersAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.filters.InternalFilters;
import org.elasticsearch.search.aggregations.bucket.filters.InternalFilters.Bucket;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.range.InternalRange;
import org.elasticsearch.search.aggregations.bucket.terms.LongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation.SingleValue;
import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
import org.elasticsearch.search.aggregations.metrics.tophits.InternalTopHits;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCount;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCountBuilder;
import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
import org.elasticsearch.search.sort.SortOrder;
import org.joda.time.DateTime;

import com.broada.apm.utils.JsonUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * 
 * 统计分析工具类
 * 详见http://www.elasticsearch.org/guide/en/elasticsearch/client/java-api/
 * current/java-aggs.html
 * 
 * @author louyi Create By 2015-1-21 上午10:56:20
 */
public class AggregationUtil {
	/**
	 * @param name 返回值name
	 * @param field 根据field进行terms
	 * @param size 返回记录数,默认10条
	 * @return
	 */
	public static TermsBuilder getTerms(String name, String field, int size) {
		return AggregationBuilders.terms(name).field(field).size(size);
	}
	/**
	 * @param name 返回值name
	 * @param field 根据field进行terms
	 * @return
	 */
	public static TermsBuilder getTerms(String name, String field) {
		return AggregationBuilders.terms(name).field(field);
	}
	/**
	 * @param name 返回值name
	 * @param field 根据field进行terms
	 * @param value 根据field的value进行terms
	 * @return
	 */
	public static FilterAggregationBuilder getFilter(String name, String field, Object value) {
		return AggregationBuilders.filter(name).filter(QueryBuilders.termQuery(field, value));
	}
	/**
	 * 过滤聚合
	 * @param name
	 * @param field
	 * @param from
	 * @param to
	 * @return
	 */
	public static FilterAggregationBuilder getRangeFilter(String name, String field, double from, double to, boolean includeLower, boolean includeUpper) {
		return AggregationBuilders.filter(name).filter(QueryBuilders.rangeQuery(field).from(from).to(to).includeLower(includeLower).includeUpper(includeUpper));
	}
	/**
	 * @param name 返回值name
	 * @param filters getFilter数组
	 * @return
	 */
	public static FiltersAggregationBuilder getFilters(String name, Map<String, QueryBuilder> filters) {
		FiltersAggregationBuilder _filterAggr = AggregationBuilders.filters(name);
		for (Map.Entry<String, QueryBuilder> entry : filters.entrySet()) {
			_filterAggr.filter(entry.getKey(), entry.getValue());
		}
		return _filterAggr;
	}
	
	/**
	 * @param name 返回值name
	 * @param filters getFilter数组
	 * @return
	 */
	public static FilterAggregationBuilder exitsFilters(String name, String field) {
			return AggregationBuilders.filter(name).filter(QueryBuilders.existsQuery(field));
	}

	/**
	 * 
	 * 
	 * @param name
	 * @param field
	 * @param sortName
	 *          注意这里不是指字段名称 而是指aggregation.name
	 * @param asc
	 *          true 为升序 false为降序
	 * @return
	 * 
	 *         example 根据response排序: TermsBuilder termsBuilder = (TermsBuilder)
	 *         AggregationUtil.getTerms("resourceId",
	 *         "resourceId","response",false);
	 *         termsBuilder.subAggregation(AggregationUtil.getSum("tx", "tx"))
	 *         .subAggregation(AggregationUtil.getSum("response", "response"))
	 *         .subAggregation(AggregationUtil.getAvg("successRate",
	 *         "successRate"))
	 *         .subAggregation(AggregationUtil.getAvg("responseTime",
	 *         "responseTime"));
	 */
	public static TermsBuilder getTerms(String name, String field, String sortName, boolean asc, int size) {
		return AggregationBuilders.terms(name).field(field).order(Terms.Order.aggregation(sortName, asc)).size(size);
	}

	/**
	 * 根据时间颗粒度统计
	 * 
	 * @param name
	 * @param field
	 * @param interval
	 *          unit:ms
	 * @return
	 */
	public static DateHistogramBuilder getDateHistogram(String name, String field, long interval) {
		return AggregationBuilders.dateHistogram(name).field(field).interval(interval);
	}
	/**
	 * @param name
	 * @param field
	 * @return
	 */
	public static AbstractAggregationBuilder getAvg(String name, String field) {
		return AggregationBuilders.avg(name).field(field);
	}
	/**
	 * script 支持结果集运算 例如_value /1000
	 * @param name
	 * @param field
	 * @param sript
	 * @return
	 */
	public static AbstractAggregationBuilder getAvg(String name, String field, Script script) {
		return AggregationBuilders.avg(name).field(field).script(script);
	}
	/**
	 * @param name
	 * @param field
	 * @return
	 */
	public static AbstractAggregationBuilder getCount(String name, String field) {
		return AggregationBuilders.count(name).field(field);
	}
	/**
	 * @param name
	 * @param field
	 * @return
	 */
	public static AbstractAggregationBuilder getMax(String name, String field) {
		return AggregationBuilders.max(name).field(field);
	}
	/**
	 * @param name
	 * @param field
	 * @return
	 */
	public static AbstractAggregationBuilder getMin(String name, String field) {
		return AggregationBuilders.min(name).field(field);
	}
	/**
	 * @param name
	 * @param field
	 * @return
	 */
	public static AbstractAggregationBuilder getSum(String name, String field) {
		return AggregationBuilders.sum(name).field(field);
	}
  /**
   * 解析时间颗粒度聚合接口
   * @param aggregations
   * @return
   */
	public static List<Map<String, Object>> parseDateHistogramAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseDateHistogram((Histogram) aggregation));
		}
		return result;
	}
	/**
	 * 解析terms聚合接口
	 * @param aggregations
	 * @return
	 */
	public static List<Map<String, Object>> parseTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseTerms((Terms) aggregation));
		}
		return result;
	}
 /**
  * 解析简单的聚合接口
  * @param aggregations
  * @return
  */
	public static Map<String, Object> parseSingleAggregation(Aggregations aggregations) {
		Map<String, Object> aggr= Maps.newHashMap();
		if (aggregations == null)
			return aggr;
		for (Aggregation aggregation : aggregations.asList()) {
			aggr.putAll(parseSingleAggregation(aggregation));
		}
		return aggr;
	}

	@SuppressWarnings("rawtypes")
	public static Map<String, Object> parseSingleAggregation(Aggregation aggregation) {
		if (aggregation == null)
			return null;
		if (aggregation instanceof InternalAvg)
			return parseInternalAvg((InternalAvg) aggregation);
		if (aggregation instanceof SingleValue)
			return parseSingle((SingleValue) aggregation);
		if (aggregation instanceof ValueCountBuilder)
			return parseCount((ValueCount) aggregation);
		if (aggregation instanceof InternalFilters)
			return parseFilters((InternalFilters) aggregation);
		if (aggregation instanceof InternalFilter)
			return parseFilter((InternalFilter) aggregation);
		if (aggregation instanceof LongTerms)
			return parseLongTerms((LongTerms) aggregation);
		if( aggregation instanceof InternalRange) {
			return parseRangeAggregtion((InternalRange) aggregation);
		}
		return null;
	}

	private static Map<String, Object> parseRangeAggregtion(@SuppressWarnings("rawtypes") InternalRange internalRange) {
		Map<String, Object> record = new HashMap<String, Object>();
		for (Object b : internalRange.getBuckets()) {
            InternalRange.Bucket rb = (InternalRange.Bucket) b;
            List<Aggregation> rbas = rb.getAggregations().asList();
            for(Aggregation rba: rbas) {
                record.putAll(parseSingleAggregation(rba));
            }
        }
		return record;
	}

	private static Map<String, Object> parseLongTerms(LongTerms aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			record.put(aggregation.getName(), bucket.getKey());
		}
		return record;
	}

	private static Map<String, Object> parseFilter(InternalFilter aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		record.put(aggregation.getName(), aggregation.getDocCount());
		return record;
	}

	private static Map<String, Object> parseFilters(InternalFilters aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		for (Bucket bucket : aggregation.getBuckets()) {
			record.put(bucket.getKey(), bucket.getDocCount());
		}
		return record;
	}

	private static Map<String, Object> parseCount(ValueCount aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		record.put(aggregation.getName(), aggregation.getValue());
		return record;
	}

	// 单独计算avg
	private static Map<String, Object> parseInternalAvg(InternalAvg aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		record.put(aggregation.getName(),Double.isNaN(aggregation.value())?null:aggregation.value());
		return record;
	}
	static Field valueFormatterField;
	static{
	    try{
        	valueFormatterField = InternalNumericMetricsAggregation.class.getDeclaredField("valueFormatter");
        	valueFormatterField.setAccessible(true);
	    } catch (Exception e) {
	        
	    }
	}
	private static Map<String, Object> parseSingle(SingleValue aggregation) {
		Map<String, Object> record = new HashMap<String, Object>();
		try{
            ValueFormatter vf = (ValueFormatter) valueFormatterField.get(aggregation);
            if(vf instanceof org.elasticsearch.search.aggregations.support.format.ValueFormatter.DateTime) {
                record.put(aggregation.getName(), new Date((long)aggregation.value()));
            } else {
                record.put(aggregation.getName(), aggregation.value() > Long.MAX_VALUE ? null : (aggregation.value() < Long.MIN_VALUE ? null : aggregation.value()));
            }
		} catch(Exception e) {
		    record.put(aggregation.getName(), aggregation.value());
		}
		return record;
	}

	private static List<Map<String, Object>> parseDateHistogram(Histogram aggregation) {
		
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for ( org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Bucket bucket : aggregation.getBuckets()) {
			Map<String, Object> record = new HashMap<String, Object>();
			// put 时间点
			//TODO  减掉8小时方法需要改进
			record.put(aggregation.getName(), ((DateTime) bucket.getKey()).getMillis());
			for (Aggregation _aggregation : bucket.getAggregations().asList()) {
				if (_aggregation instanceof StringTerms) {
					List<Map<String, Object>> terms = parseTerms((Terms) _aggregation);
					for (int i = 0; i < terms.size(); i++) {
						record.putAll(terms.get(i));
					}
				} else
					record.putAll(parseSingleAggregation(_aggregation));
			}
			result.add(record);
		}
		return result;
	}

	private static List<Map<String, Object>> parseTerms(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			Map<String, Object> record = new HashMap<String, Object>();
			record.put(aggregation.getName(), bucket.getKey());
			for (Aggregation _aggregation : bucket.getAggregations().asList()) {
				if (_aggregation instanceof StringTerms) {
					List<Map<String, Object>> terms = parseTerms((Terms) _aggregation);
					for (int i = 0; i < terms.size(); i++) {
						record.putAll(terms.get(i));
					}
				} else
					record.putAll(parseSingleAggregation(_aggregation));
			}
			result.add(record);
		}
		return result;
	}

	private static List<Map<String, Object>> parseMultiTerms(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			parseMultiTerms(bucket, aggregation, null, result);
		}
		return result;
	}

	private static void parseMultiTerms(Terms.Bucket bucket, Terms aggregation, Map<String, Object> parent,
			List<Map<String, Object>> result) {
		List<Terms> list = new ArrayList<Terms>();
		Map<String, Object> record = new HashMap<String, Object>();
		record.put(aggregation.getName(), bucket.getKey());
		if (parent != null)
			record.putAll(parent);
		for (Aggregation _aggregation : bucket.getAggregations().asList()) {
			if (_aggregation instanceof Terms) {
				list.add((Terms) _aggregation);
			} else {
				record.putAll(parseSingleAggregation(_aggregation));
			}
		}
		for (Terms subTerm : list) {
			List<Terms.Bucket> subBuckets = subTerm.getBuckets();
			if (subBuckets != null && !subBuckets.isEmpty()) {
				for (Terms.Bucket bucket2 : subBuckets) {
					parseMultiTerms(bucket2, subTerm, record, result);
				}
				return ;
			}
		}
		result.add(record);
	}
	
	public static List<Map<String, Object>> parseNestedMultiTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseNestedMultiTerms((Terms) aggregation));
		}
		return result;
	}
	
	private static List<Map<String, Object>> parseNestedMultiTerms(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			parseNestedMultiTerms(bucket, aggregation, null, result);
		}
		return result;
	}

	private static void parseNestedMultiTerms(Terms.Bucket bucket, Terms aggregation, Map<String, Object> parent,
			List<Map<String, Object>> result) {
		List<Terms> list = new ArrayList<Terms>();
		Map<String, Object> record = new HashMap<String, Object>();
		record.put(aggregation.getName(), bucket.getKey());
		if (parent != null)
			parent.put(bucket.getKey().toString(), record);
		for (Aggregation _aggregation : bucket.getAggregations().asList()) {
			if (_aggregation instanceof Terms) {
				list.add((Terms) _aggregation);
			} else {
				record.putAll(parseSingleAggregation(_aggregation));
			}
		}
		for (Terms subTerm : list) {
			List<Terms.Bucket> subBuckets = subTerm.getBuckets();
			if (subBuckets != null && !subBuckets.isEmpty()) {
				for (Terms.Bucket bucket2 : subBuckets) {
					parseMultiTerms(bucket2, subTerm, record, result);
				}
				return ;
			}
		}
		result.add(record);
	}
	
	/**
	 * 解析terms聚合接口,适用于多字段分组情景
	 * 
	 * @param aggregations
	 * @return
	 */
	public static List<Map<String, Object>> parseMultiTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseMultiTerms((Terms) aggregation));
		}
		return result;
	}
	
	/**
	 * 获取tophits
	 * @param aggregations
	 * @return
	 */
	public static List<Map<String, Object>> parseTopHitsTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseTopHitsTerms((Terms) aggregation));
		}
		return result;
		
	}
	
	private static List<Map<String, Object>> parseTopHitsTerms(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			Map<String, Object> record = new HashMap<String, Object>();
			record.put(aggregation.getName(), bucket.getKey());
			
			TopDocs[] shardDocs = new TopDocs[bucket.getAggregations().asList().size()];
			if (bucket.getAggregations().asList().size() > 0) {
				for (int i = 0; i < shardDocs.length; i++) {
					InternalTopHits topHitsAgg = (InternalTopHits) bucket.getAggregations().asList().get(i);
					for (SearchHit hit : topHitsAgg.getHits()) {
						String source = hit.sourceAsString();
						try {
							Map<String, Object> map = JsonUtil.json2map(source);
							record.putAll(map);
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			}
			result.add(record);
		}
		return result;
		
	}
	
	/**
	 * @param name
	 * @return
	 */
	public static AbstractAggregationBuilder getTopHits(String name, String field, SortOrder order, int size, String[] includes, String[] excludes) {
		return AggregationBuilders.topHits(name).addSort(field, order).setSize(size).setFetchSource(includes, excludes);
	}
	
	public static List<Map<String, Object>> parseTradeTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseTradeTerms((Terms) aggregation));
		}
		return result;
	}
	
	private static List<Map<String, Object>> parseTradeTerms(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			Map<String, Object> record = new HashMap<String, Object>();
			record.put(aggregation.getName(), bucket.getKey());
			for (Aggregation _aggregation : bucket.getAggregations().asList()) {
				if (_aggregation instanceof StringTerms) {
					List<Map<String, Object>> terms = parseTerms((Terms) _aggregation);
					for (int i = 0; i < terms.size(); i++) {
						for (String key : terms.get(i).keySet()) {
							if (record.get(key) != null) {
								record.put(key, record.get(key) + "," + terms.get(i).get(key));
							} else {
								record.put(key, terms.get(i).get(key));
							}
						}
					}
				} else
					record.putAll(parseSingleAggregation(_aggregation));
			}
			result.add(record);
		}
		return result;
	}
	
	public static List<Map<String, Object>> parseSubTermsAggregations(Aggregations aggregations) {
		List<Map<String, Object>> result = Lists.newArrayList();
		if (aggregations == null)
			return result;
		for (Aggregation aggregation : aggregations.asList()) {
			result.addAll(parseSubTermAggs((Terms) aggregation));
		}
		return result;
	}
	
	private static List<Map<String, Object>> parseSubTermAggs(Terms aggregation) {
		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		for (Terms.Bucket bucket : aggregation.getBuckets()) {
			for (Aggregation _aggregation : bucket.getAggregations().asList()) {
				if (_aggregation instanceof StringTerms) {
//					List<Map<String, Object>> terms = parseTerms((Terms) _aggregation);
//					for (int i = 0; i < terms.size(); i++) {
//						for (String key : terms.get(i).keySet()) {
//							if (record.get(key) != null) {
//								record.put(key, record.get(key) + "," + terms.get(i).get(key));
//							} else {
//								record.put(key, terms.get(i).get(key));
//							}
//						}
//					}
					
					
					for (Terms.Bucket subBucket : ((Terms) _aggregation).getBuckets()) {
						Map<String, Object> record = new HashMap<String, Object>();
						record.put(aggregation.getName(), bucket.getKey()); // typeName
						record.put(_aggregation.getName(), subBucket.getKey()); // serviceId
						for (Aggregation _subAggregation : subBucket.getAggregations().asList()) {
							if (_subAggregation instanceof StringTerms) {
								List<Map<String, Object>> terms = parseTerms((Terms) _subAggregation);
								record.put("channel", terms);
							} else {
								record.putAll(parseSingleAggregation(_subAggregation));
							}
						}
						result.add(record);
					}
				} else {
					
				}
//					record.putAll(parseSingleAggregation(_aggregation));
			}
		}
		return result;
	}
}